package main

import (
	"fmt"
	"sync"
	"time"

	consulapi "github.com/hashicorp/consul/api"
	"github.com/hashicorp/consul/api/watch"
)

// watch包的使用方法为：1）使用watch.Parse(查询参数)生成Plan，2）绑定Plan的handler，3）运行Plan

// Watcher 定义watcher
type Watcher struct {
	Address    string                 // consul agent 的地址："127.0.0.1:8500"
	Wp         *watch.Plan            // 总的Services变化对应的Plan
	watchers   map[string]*watch.Plan // 对已经进行监控的service作个记录
	ServiceMap map[string]struct{}    // 这里直接用redis的map
	RWMutex    *sync.RWMutex
}

func (w *Watcher) Handler(workerID string) {
	fmt.Println(workerID, "handler执行了")
	<-time.After(20 * time.Second) // 等待20s执行:健康检查5s+撤销服务时间15s
	w.RWMutex.RLock()
	defer w.RWMutex.RUnlock()
	_, ok := w.ServiceMap[workerID]
	if !ok {
		fmt.Println(workerID, "没有这个key了")
		return
	}
	delete(w.ServiceMap, workerID)
	// TODO 归还workerID
	fmt.Println("归还" + workerID + "的workerID")
}

// 将consul新增的service加入，并监控
func (w *Watcher) registerServiceWatcher(serviceName string) error {
	// watch endpoint 的请求参数，具体见官方文档：https://www.consul.io/docs/dynamic-app-config/watches#service
	wp, err := watch.Parse(map[string]interface{}{
		"type":    "service",
		"service": serviceName,
	})
	if err != nil {
		return err
	}

	// 定义service变化后所执行的程序(函数)handler
	wp.Handler = func(idx uint64, data interface{}) {
		switch d := data.(type) {
		case []*consulapi.ServiceEntry:
			for _, i := range d {
				// 这里是单个service变化时需要做的逻辑，可以自己添加，或在外部写一个类似handler的函数传进来
				// fmt.Printf("service %s 已变化", i.Service.ID)
				// 打印service的状态
				// fmt.Println("service status: ", i.Checks.AggregatedStatus())

				fmt.Printf("service %s 已变化; status: %s; serviceMap: %+v\n", i.Service.ID, i.Checks.AggregatedStatus(), w.ServiceMap)
				switch i.Checks.AggregatedStatus() {
				case consulapi.HealthCritical: // 健康检查出错
					fmt.Println(i.Service.ID, "出错了")
					w.RWMutex.Lock()
					w.ServiceMap[i.Service.ID] = struct{}{}
					w.RWMutex.Unlock()
					go w.Handler(i.Service.ID)
				case consulapi.HealthPassing: // 健康检查通过
					w.RWMutex.Lock()
					delete(w.ServiceMap, i.Service.ID)
					w.RWMutex.Unlock()
				}
			}
		}
	}
	// 启动监控
	go func() {
		_ = wp.Run(w.Address)
	}()
	// 对已启动监控的service作一个记录
	w.RWMutex.Lock()
	w.watchers[serviceName] = wp
	w.RWMutex.Unlock()

	return nil
}

func NewWatcher(watchType string, opts map[string]string, consulAddr string) (*Watcher, error) {
	var options = map[string]interface{}{
		"type": watchType,
	}
	// 组装请求参数。(监控类型不同，其请求参数不同)
	for k, v := range opts {
		options[k] = v
	}

	wp, err := watch.Parse(options)
	if err != nil {
		return nil, err
	}

	w := &Watcher{
		Address:    consulAddr,
		Wp:         wp,
		watchers:   make(map[string]*watch.Plan),
		ServiceMap: make(map[string]struct{}),
		RWMutex:    new(sync.RWMutex),
	}

	wp.Handler = func(idx uint64, data interface{}) {
		switch d := data.(type) {
		// 这里只实现了对services的监控，其他监控的data类型判断参考：https://github.com/dmcsorley/avast/blob/master/consul.go
		// services
		case map[string][]string:
			for i := range d {
				// 如果该service已经加入到ConsulRegistry的services里监控了，就不再加入 或者i 为 "consul"的字符串
				// 为什么会多一个consul，参考官方文档services监听的返回值：https://www.consul.io/docs/dynamic-app-config/watches#services
				if _, ok := w.watchers[i]; ok || i == "consul" {
					continue
				}
				_ = w.registerServiceWatcher(i)
			}

			// 从总的services变化中找到不再监控的service并停止
			w.RWMutex.RLock()
			watches := w.watchers
			w.RWMutex.RUnlock()

			// remove unknown services from watchers
			for i, svc := range watches {
				if _, ok := d[i]; !ok {
					svc.Stop()
					delete(watches, i)
				}
			}
		default:
			fmt.Printf("不能判断监控的数据类型: %v", &d)
		}
	}

	return w, nil
}

func RegisterWatcher(watchType string, opts map[string]string, consulAddr string) error {
	w, err := NewWatcher(watchType, opts, consulAddr)
	if err != nil {
		fmt.Println(err)
		return err
	}
	defer w.Wp.Stop()
	if err = w.Wp.Run(consulAddr); err != nil {
		fmt.Println("err: ", err)
		return err
	}

	return nil
}

func main() {
	// address := "127.0.0.1:8500"
	address := "192.168.1.2:8500"
	if err := RegisterWatcher("services", nil, address); err != nil {
		fmt.Println("启动 consul 的watch监控失败")
	}
}
